import bean.Event;
import bean.Student;
import bean.Teacher;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple5;
import org.apache.flink.api.java.typeutils.PojoTypeInfo;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.InstantiationUtil;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.Serializable;
import java.time.Duration;
import java.util.List;
import java.util.Properties;
import java.util.Scanner;
import java.util.function.Function;

/**
 * connect之后生成ConnectedStreams，会对两个流的数据应用不同的处理方法，并且双流 之间可以共享状态(比如计数)。
 * 这在第一个流的输入会影响第二个流 时, 会非常有用;
 * <p>
 * union是DataStream* → DataStream。
 * <p>
 * connect只能连接两个流，而union可以连接多于两个流 。
 * <p>
 * connect连接的两个流类型可以不一致，而union连接的流的类型必须一致。
 */
public class TransformTest4_MultipleStreamsConverge {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        sEnv.setParallelism(1);

        //InstantiationUtil.serializeObject(func);

        DataStream<Student> student = sEnv.fromCollection(InitData.loadData());
        DataStream<Teacher> teacher = sEnv.fromCollection(InitData.loadData1());

        //采用connect连接两个stream.
        ConnectedStreams<Student, Teacher> connect = student.connect(teacher);
        connect.process(new CoProcessFunction<Student, Teacher, Tuple5<String, Integer, String, String, Long>>() {
            @Override
            public void processElement1(Student value, Context ctx, Collector<Tuple5<String, Integer, String, String, Long>> out) throws Exception {
                out.collect(InitData.studentTuple5(value));
            }

            @Override
            public void processElement2(Teacher value, Context ctx, Collector<Tuple5<String, Integer, String, String, Long>> out) throws Exception {
                out.collect(InitData.teacherTuple5(value));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.INT, Types.STRING, Types.STRING, Types.LONG))
          .print("process");

        // connect
        connect.map(new CoMapFunction<Student, Teacher, Tuple5<String, Integer, String, String, Long>>() {
            @Override
            public Tuple5<String, Integer, String, String, Long> map1(Student value) throws Exception {
                 return InitData.studentTuple5(value);
            }

            @Override
            public Tuple5<String, Integer, String, String, Long> map2(Teacher value) throws Exception {
                return InitData.teacherTuple5(value);
            }
        }).filter(new FilterFunction<Tuple5<String, Integer, String, String, Long>>() {
                    @Override
                    public boolean filter(Tuple5<String, Integer, String, String, Long> value) throws Exception {
                        return value.f1 > 1000;
                    }
                }).returns(Types.TUPLE(Types.STRING, Types.INT, Types.STRING, Types.STRING, Types.LONG))
          .print("map");

        sEnv.execute("ConnectOperator");

        UnionTest.test();
    }


    static class UnionTest{
        static void test() throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(1);
            // 登录日志流
            DataStream<Event> loginEventStream = env.fromElements(
                    new Event(1659657901L, "A", "192.168.1.1", "0", "2022-08-05 08:05:01", null, null),
                    new Event(1659657903L, "A", "192.168.1.1", "0", "2022-08-05 08:05:03", null, null),
                    new Event(1659657905L, "A", "192.168.1.1", "0", "2022-08-05 08:05:05", null, null),
                    new Event(1659657907L, "A", "192.168.1.1", "0", "2022-08-05 08:05:07", null, null)
            );
            // 下载日志流
            DataStream<Event> downloadEventStream = env.fromElements(
                    new Event(1659657902L, "A", null, null, "2022-08-05 08:05:02", "O", "西游记"),
                    new Event(1659657904L, "A", null, null, "2022-08-05 08:05:04", "1", "三国志"),
                    new Event(1659657906L, "A", null, null, "2022-08-05 08:05:06", "1", "红楼梦"),
                    new Event(1659657908L, "A", null, null, "2022-08-05 08:05:08", "1", "水浒传"),
                    new Event(1659657966L, "A", null, null, "2022-08-05 08:06:06", "1", "山海经")
            );
            // 双流联合（union） 且 控制台打印
            loginEventStream.union(downloadEventStream)
                    .print();
            // 启动任务
            env.execute("LoginUnionDownloadDataStreamTest");
        }
     }


     public static class UnionTest1{
         public static void main() throws Exception {
             // 创建环境
             StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
             // 登录日志流
             DataStream<Event> loginEventStream = env.fromElements(
                     new Event(1659657901L,"2022-08-05 08:05:01", "192.168.1.1", "0",  null, null, null),
                     new Event(1659657903L,"2022-08-05 08:05:03", "192.168.1.1", "0",  null, null, null),
                     new Event(1659657905L, "2022-08-05 08:05:05", "192.168.1.1", "0", null, null, null),
                     new Event(1659657907L, "2022-08-05 08:05:07", "192.168.1.1", "0", null, null, null)
             ).assignTimestampsAndWatermarks(
                     WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(1L))
                             .withTimestampAssigner(
                                     (Event event, long l) -> {
                                         return event.getTimestamp();
                                     }
                             )
             );
             // 下载日志流
             DataStream<Event> downloadEventStream = env.fromElements(
                     new Event(1659657902L,"2022-08-05 08:05:02", null, null,  null, "O", "西游记"),
                     new Event(1659657904L,"2022-08-05 08:05:04", null, null,  null, "1", "三国志"),
                     new Event(1659657906L,"2022-08-05 08:05:06", null, null,  null, "1", "红楼梦"),
                     new Event(1659657908L, "2022-08-05 08:05:08", null, null, null, "1", "水浒传"),
                     new Event(1659657966L, "2022-08-05 08:06:06", null, null, null, "1", "山海经")
             ).assignTimestampsAndWatermarks(
                     WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(1L))
                             .withTimestampAssigner(
                                     (Event event, long l) -> {
                                         return event.getTimestamp();
                                     }
                             )
             );
             // 联合（union） 数据流，并按照时间戳排序且打印输出
             loginEventStream.union(downloadEventStream)
                     .keyBy(r -> r.getKey())
                     .process(new SortByTimestampFunction())
                     .print();
             // 启动任务
             env.execute("UnionAndSortDataStreamTest");
         }
     }

    public static class InitData{
        private static ObjectMapper mapper = new ObjectMapper();

        public static String inputPath = "./flink-aword-count-test/src/main/resources/std.json";
        public static List<Student> loadData() throws FileNotFoundException, JsonProcessingException {
            Scanner scanner = new Scanner(new File(inputPath));
            List<Student> list = Lists.newArrayList();

            while (scanner.hasNextLine()){
                var json = scanner.nextLine();
                Student student = mapper.readValue(json, new TypeReference<Student>(){});
                list.add(student);
            }

            return list;
        }

        public static String inputPath1 = "./flink-aword-count-test/src/main/resources/std.json";
        public static <T> List<T> loadData1() throws FileNotFoundException, JsonProcessingException {
            Scanner scanner = new Scanner(new File(inputPath1));
            List<T> list = Lists.newArrayList();

            while (scanner.hasNextLine()){
                var json = scanner.nextLine();
                T teacher = mapper.readValue(json, new TypeReference<T>(){});
                list.add(teacher);
            }

            return list;
        }

        static Tuple5<String, Integer, String, String, Long> studentTuple5(Student value1){
            return new Tuple5<>(value1.name, value1.age, value1.sex, value1.classId, value1.timestamp);
        }

        static Tuple5<String, Integer, String, String, Long> teacherTuple5(Teacher value1){
            return new Tuple5<>(value1.name, value1.age, value1.sex, value1.classId, value1.timestamp);
        }
    }
}

